package Example

import org.apache.spark.sql.SparkSession

object preprocess {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .master("local[*]")
      .appName("Data Explore")
      //设置Hive连接参数
      .config("hive.metastore.uris", "thrift://master:9083")
      .config("spark.sql.warehouse.dir", "hdfs://master:8020/user/hive/warehouse")
      .enableHiveSupport()
      .getOrCreate()

    val media_index = spark.read.table("portrait.media_index")
    val billevent = spark.read.table("portrait.billevent")
    val orderData = spark.read.table("portrait.orderData")
    val usermsg = spark.read.table("portrait.usermsg")
    val userevent = spark.read.table("portrait.userevent")

    val mediaData_process = media_index.distinct()
      .filter("owner_name not like '%EA%' and owner_name not like '%EB%' and owner_name not like '%EC%' and owner_name not like '%ED%' and owner_name not like '%EE%'")
      .filter("owner_code !='02' and owner_code !='09' and owner_code != '10'")
      .filter("duration >=20000 and duration <=5*1000*60*60")
      .filter("sm_name =='珠江宽频' or sm_name =='互动电视' or sm_name == '甜果电视' or sm_name == '数字电视'")
      .filter("res_type !=0 or origin_time not like '%00' or end_time not like '%00'")

    val usereventData_process = userevent.distinct()
      .filter("owner_name not like '%EA%' and owner_name not like '%EB%' and owner_name not like '%EC%' and owner_name not like '%ED%' and owner_name not like '%EE%'")
      .filter("owner_code !='02' and owner_code !='09' and owner_code != '10'")
      .filter("run_name == '正常' or run_name == '主动暂停' or run_name =='欠费暂停' or run_name=='主动销户'")

    val usermsgData_process = usermsg.distinct()
      .filter("owner_name not like '%EA%' and owner_name not like '%EB%' and owner_name not like '%EC%' and owner_name not like '%ED%' and owner_name not like '%EE%'")
      .filter("owner_code !='02' and owner_code !='09' and owner_code != '10'")
      .filter("sm_name =='珠江宽频' or sm_name =='互动电视' or sm_name == '甜果电视' or sm_name == '数字电视'")
      .filter("run_name == '正常' or run_name == '主动暂停' or run_name =='欠费暂停' or run_name=='主动销户'")

    val billeventsData_process = billevent.distinct()
      .filter("owner_name not like '%EA%' and owner_name not like '%EB%' and owner_name not like '%EC%' and owner_name not like '%ED%' and owner_name not like '%EE%'")
      .filter("owner_code !='02' and owner_code !='09' and owner_code != '10'")
      .filter("sm_name =='珠江宽频' or sm_name =='互动电视' or sm_name == '甜果电视' or sm_name == '数字电视'")

    val orderData_process = orderData.distinct().filter("owner_name not like '%EA%' and owner_name not like '%EB%' and owner_name not like '%EC%' and owner_name not like '%ED%' and owner_name not like '%EE%'")
      .filter("owner_code !='02' and owner_code !='09' and owner_code != '10'")
      .filter("sm_name =='珠江宽频' or sm_name =='互动电视' or sm_name == '甜果电视' or sm_name == '数字电视'")
      .filter("run_name == '正常' or run_name == '主动暂停' or run_name =='欠费暂停' or run_name=='主动销户'")

    //将预处理完的数据存入hive表中
    mediaData_process.write.mode("overwrite").saveAsTable("portrait.media_index_process")
    orderData_process.write.mode("overwrite").saveAsTable("portrait.order_index_process")
    usermsgData_process.write.mode("overwrite").saveAsTable("portrait.mediamatch_usermsg_process")
    usereventData_process.write.mode("overwrite").saveAsTable("portrait.mediamatch_userevent_process")
    billeventsData_process.write.mode("overwrite").saveAsTable("portrait.mmconsume_billevents_process")

    //用户基本信息表数据预处理验证
    //    originUsermsg = spark.sql("select * from portrait.mediamatch_usermsg")
    //    processedUsermsg = spark.sql("select * from  portrait.mediamatch_usermsg_process")
    //    originUsermsg.count()
    //    processedUsermsg.count()
    //    processedUsermsg.select("phone_no").distinct().count()
    //    processedUsermsg.select("sm_name").distinct().show()
    //    processedUsermsg.select("run_name").distinct().show()
    //    processedUsermsg.select("owner_name").distinct().show()
    //    processedUsermsg.select("owner_code").distinct().show()

  }
}
